package com.etlmaster.datax.plugins.writer.hivewriter;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.sql.SQLException;
import java.text.MessageFormat;

import org.apache.commons.codec.digest.DigestUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.util.Shell.ShellCommandExecutor;

import com.dp.nebula.wormhole.common.JobStatus;
import com.dp.nebula.wormhole.common.WormholeException;
import com.dp.nebula.wormhole.common.interfaces.IParam;
import com.dp.nebula.wormhole.common.interfaces.ISourceCounter;
import com.dp.nebula.wormhole.common.interfaces.ITargetCounter;
import com.dp.nebula.wormhole.common.interfaces.IWriterPeriphery;
import com.etlmaster.datax.plugins.common.DFSUtils;
import com.etlmaster.datax.plugins.reader.hivereader.HiveJdbcClient;
import com.etlmaster.executor.utils.DataCollectionUtils;
import com.etlmaster.executor.utils.LogWriter;
import com.hadoop.compression.lzo.LzoIndexer;

public class HiveWriterPeriphery implements IWriterPeriphery {

	private static final int HIVE_TABLE_ADD_PARTITION_PARAM_NUMBER = 2;
	private static final String HIDDEN_FILE_PREFIX = "_";
	private static final int MAX_LZO_CREATION_TRY_TIMES = 3;
	private static final long LZO_CREATION_TRY_INTERVAL_IN_MILLIS = 10000L;

	private String dir = "";
	private String prefixname = "";
	private int concurrency = 1;
	private String codecClass = "";
	private String fileType = "TXT";
	private String suffix = "";
	private boolean lzoCompressed = false;
	private String hiveTableAddPartitionOrNot = "false";
	private String hiveTableAddPartitionCondition = "";

	private final String ADD_PARTITION_SQL = "alter table {0} add if not exists partition({1}) location ''{2}'';";

	private FileSystem fs;
	private LzoIndexer lzoIndexer;
	
	private HiveJdbcClient client;
	private String tempdir;
	private Configuration conf;
	private String hiveUrl="jdbc:hive2://192.168.230.131:10000/dc";
	private String username="hadoop";
	private String password="hadoop";
	private String hiveTableName="inter_file_log";
	private final String LOAD_HIVE_SQL = DataCollectionUtils.getProperties("LOAD_HIVE_SQL");
	private final String LOAD_HIVE_PART_SQL = DataCollectionUtils.getProperties("LOAD_HIVE_PART_SQL");
	private Path absolutePath;
	private String partWhere="";
	public void rollback(IParam param) {
		deleteFilesOnHdfs(this.dir, this.prefixname, suffix);
	}

	
	/**
	 * 事后处理。重命名文件
	 * 将temp目录下的文件移植到hive的表中，如果是分区表则将数据一张到分区中
	 */
	public void doPost(IParam param, ITargetCounter counter) {
		//deleteFilesOnHdfs(this.dir, this.prefixname, suffix);
		// rename temp files, make them visible to hdfs
		
		
		
		try {

			//renameFiles();
			renameFiles();

			URI uri = new URI(tempdir);
			//System.out.println(uri.getPath());
			String sql="";
			boolean partFlag=false;
			
			if(partWhere.trim().length()==0)
			{
				sql=MessageFormat.format(LOAD_HIVE_SQL,
						"'"+uri.getPath()+"/*'", hiveTableName);
			}
			else
			{	sql=MessageFormat.format(LOAD_HIVE_PART_SQL,
						"'"+uri.getPath()+"/*'", hiveTableName,partWhere);
				partFlag=true;
			}
			LogWriter.addLog("DEBUG",hiveUrl);
			client = new HiveJdbcClient.Builder(hiveUrl).username(username)
					.password(password).sql(sql).build();
			client.initialize();
			client.processExecuteSql(sql,partFlag);
			
			
			fs = DFSUtils.createFileSystem2(new URI(tempdir),
					DFSUtils.getConf2(tempdir, null));
			if (fs.exists(absolutePath)) {
				fs.delete(absolutePath, true);
				LogWriter.addLog("INFO",absolutePath.toString()
						+ " has been deleted at dopost stage");
			}
			
			
		} catch (IOException e) {
			LogWriter.addLog("INFO","***************************************************");
			LogWriter.addLog("ERROR","delete hdfs temp folder fail:"+absolutePath.toString()+e.getMessage());
			LogWriter.addLog("INFO","***************************************************");
		}catch(URISyntaxException e){
			LogWriter.addLog("ERROR","***************************************************");
			LogWriter.addLog("ERROR","hdfs temp folder error:"+absolutePath.toString());
			LogWriter.addLog("ERROR","***************************************************");
		}catch(SQLException e){
			LogWriter.addLog("ERROR","***************************************************");
			LogWriter.addLog("ERROR","*load hive table "+hiveTableName +" failed!!!!"+e.getMessage());
			LogWriter.addLog("ERROR","***************************************************");
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}finally{
			closeAll();
		}
		
		if (lzoCompressed
				&& "true".equalsIgnoreCase(param.getValue(
						ParamKey.createLzoIndexFile, "true").trim())) {
			createLzoIndex(dir);
		}

		/* add hive table partition if necessary */
		hiveTableAddPartitionOrNot = param.getValue(
				ParamKey.hiveTableAddPartitionSwitch, "false").trim();
		if (hiveTableAddPartitionOrNot.equalsIgnoreCase("true")) {
			hiveTableAddPartitionCondition = param.getValue(
					ParamKey.hiveTableAddPartitionCondition,
					this.hiveTableAddPartitionCondition);

			if (!StringUtils.isBlank(hiveTableAddPartitionCondition)) {
				String[] hqlParam = StringUtils.split(
						hiveTableAddPartitionCondition, '@');
				
				//hiveTableAddPartitionCondition dt='2010-01-01'@sampleDatabase.sampleTable
				if (HIVE_TABLE_ADD_PARTITION_PARAM_NUMBER == hqlParam.length) {
					String parCondition = hqlParam[0].trim().replace('"', '\'');
					String uri = hqlParam[1].trim();//sampleDatabase.sampleTable
					// split dbname and tablename
					String[] parts = StringUtils.split(uri, '.');

					if (StringUtils.isBlank(parCondition)
							|| StringUtils.isBlank(uri) || parts.length != 2
							|| StringUtils.isBlank(parts[0])
							|| StringUtils.isBlank(parts[1])) {
						LogWriter.addLog("ERROR",ParamKey.hiveTableAddPartitionSwitch
								+ " param can not be parsed correctly, please check it again");
						return;
					}

					try {                    //sampleDatabase  sampleTable
						addHiveTablePartition(parts[0].trim(), parts[1].trim(),
								parCondition, dir);
					} catch (IOException e) {
						LogWriter.addLog("ERROR","add hive table partition failed:"
								+ e.getMessage());
					}
				}
			}
		}
		return;
	}

	private void renameFiles() {
		try {
			String tempPrefixnamex = HIDDEN_FILE_PREFIX + prefixname;

			fs = DFSUtils.createFileSystem2(new URI(tempdir),
					DFSUtils.getConf2(tempdir, null));
			for (FileStatus status : fs.listStatus(new Path(tempdir))) {
				if (!status.isDir()) {
					String fileName = status.getPath().getName();
					if (fileName.startsWith(tempPrefixnamex)) {
						String parentPathName = status.getPath().getParent()
								.toString();
						Path absoluteSrcPath = status.getPath();
						// trim the ahead one '_' character
						fileName = fileName.substring(1);
						Path absoluteDstPath = new Path(parentPathName + "/"
								+ fileName);
						if (fs.rename(absoluteSrcPath, absoluteDstPath)) {
							LogWriter.addLog("INFO","successfully rename file from "
									+ absoluteSrcPath.toString() + " to "
									+ absoluteDstPath.toString());
						} else {
							LogWriter.addLog("ERROR","failed to rename file from "
									+ absoluteSrcPath.toString() + " to "
									+ absoluteDstPath.toString());
						}
					}
				}
			}
		} catch (Exception e) {
			LogWriter.addLog("ERROR",String.format(
					"HdfsWriter rename temp files failed :%s,%s",
					e.getMessage(), e.getCause()));
			throw new WormholeException(
					"HdfsWriter rename temp files failed in do-post stage",
					JobStatus.POST_WRITE_FAILED.getStatus());
		} finally {
			closeAll();
		}
	}

	private void addHiveTablePartition(String dbName, String tableName,
			String partitionCondition, String location) throws IOException {
		StringBuffer addParitionCommand = new StringBuffer();

		addParitionCommand.append("hive -e \"");
		addParitionCommand.append("use " + dbName + ";");
		addParitionCommand.append(MessageFormat.format(ADD_PARTITION_SQL,
				tableName, partitionCondition, location));
		addParitionCommand.append("\"");

		LogWriter.addLog("DEBUG",addParitionCommand.toString());

		String[] shellCmd = { "bash", "-c", addParitionCommand.toString() };
		ShellCommandExecutor shexec = new ShellCommandExecutor(shellCmd);
		shexec.execute();
		int exitCode = shexec.getExitCode();
		if (exitCode != 0) {
			throw new IOException("process exited with exit code " + exitCode);
		} else {
			LogWriter.addLog("INFO","hive table add partion hql executed correctly:"
					+ addParitionCommand.toString());
		}
	}

	// private void createHiveTable(IParam param){
	// String createTableOrNot = param.getValue(
	// ParamKey.hiveTableSwitch, "false");
	//
	// if (!createTableOrNot.equalsIgnoreCase("true"))
	// return;
	//
	// String tableFieldType = param.getValue(ParamKey.tableFieldType,
	// "").trim().toUpperCase();
	// if (StringUtils.isEmpty(tableFieldType))
	// return;
	//
	// String[] fieldTypes = tableFieldType.split(",");
	// for (int i = 0; i < fieldTypes.length; i++) {
	// String fieldType = fieldTypes[i];
	// if (fieldType.equalsIgnoreCase("bigint")){
	//
	// }else if(fieldType.equalsIgnoreCase("string")){
	//
	// }else{
	// logger.error(String.format("unnormal fieldType:%s in the param %s",
	// tableFieldType, ParamKey.tableFieldType));
	// return;
	// }
	// }
	// }

	private void createLzoIndex(String directory) {
		try {
			Configuration cfg = DFSUtils.getConf(directory, null);
			lzoIndexer = new LzoIndexer(cfg);

			int times = 1;
			boolean idxCreated = false;
			do {
				LogWriter.addLog("INFO","start to create lzo index file on " + directory
						+ " times: " + times);
				try {
					lzoIndexer.index(new Path(directory));
					idxCreated = true;
				} catch (Throwable t) {
					LogWriter.addLog("ERROR",String
							.format("HdfsWriter doPost stage create index %s failed, start to sleep %d millis sec, %s,%s",
									directory,
									LZO_CREATION_TRY_INTERVAL_IN_MILLIS,
									t.getMessage(), t.getCause()));
					try {
						Thread.sleep(LZO_CREATION_TRY_INTERVAL_IN_MILLIS);
					} catch (InterruptedException ite) {
						ite.printStackTrace(System.err);
					}
				}
			} while (!idxCreated && times++ <= MAX_LZO_CREATION_TRY_TIMES);

			if (!idxCreated) {
				throw new WormholeException(
						"lzo index creation failed after try "
								+ MAX_LZO_CREATION_TRY_TIMES + " times",
						JobStatus.POST_WRITE_FAILED.getStatus());
			}

			if (existIncompleteFile(dir)) {
				LogWriter.addLog("ERROR",String.format(
						"HdfsWriter doPost stage create index %s failed:",
						directory));
				throw new WormholeException(
						"dfsWriter doPost stage create index failed",
						JobStatus.POST_WRITE_FAILED.getStatus());
			}
		} catch (IOException e) {
			LogWriter.addLog("ERROR",String.format(
					"HdfsWriter doPost stage get configuration failed:%s,%s",
					e.getMessage(), e.getCause()));
			throw new WormholeException(e,
					JobStatus.POST_WRITE_FAILED.getStatus());
		}
	}

	public void prepare(IParam param, ISourceCounter counter) {
		dir = param.getValue(ParamKey.dir);
		partWhere = param.getValue(ParamKey.partWhere);
		prefixname = param.getValue(ParamKey.prefixname, "prefix");
		concurrency = param.getIntValue(ParamKey.concurrency, 1);

		if (dir.endsWith("*")) {
			dir = dir.substring(0, dir.lastIndexOf("*"));
		}
		if (dir.endsWith("/")) {
			dir = dir.substring(0, dir.lastIndexOf("/"));
		}
		
		try {
			createTempDir();
			//param.putValue(ParamKey.tempdir, absolutePath.toString());
			tempdir=absolutePath.toString();
			param.putValue(ParamKey.tempdir, tempdir);
			//String aaa=absolutePath.toString();
		} catch (Exception e) {
			throw new WormholeException(e,
					JobStatus.READ_FAILED.getStatus());
		}
		

		fileType = param.getValue(ParamKey.fileType, this.fileType);
		codecClass = param.getValue(ParamKey.codecClass, "");
		if (fileType.equalsIgnoreCase("TXT_COMP")) {
			suffix = DFSUtils.getCompressionSuffixMap().get(codecClass);
			if (StringUtils.isEmpty(suffix)) {
				suffix = "lzo";
			}
		}

		if (suffix.equalsIgnoreCase("lzo")) {
			lzoCompressed = true;
		}
		
		
		hiveUrl=param.getValue(ParamKey.url, "");
		username=param.getValue(ParamKey.username, "");
		password=param.getValue(ParamKey.password, "");
		hiveTableName=param.getValue(ParamKey.hiveTableName, "");
		
		deleteFilesOnHdfs(this.dir, HIDDEN_FILE_PREFIX + this.prefixname,
				suffix);
	}

	private String createFilename(String sql) {
		return DigestUtils.md5Hex(sql + System.currentTimeMillis());
	}
	/**
	 * 在dir目录下新建临时文件夹，作为导出数据的目录
	 * @return
	 * @throws Exception
	 */
	private String createTempDir() throws Exception {
		//conf = DFSUtils.getConf(dataDir, null);
		conf = DFSUtils.getConf2(this.dir, null); 
		fs = DFSUtils.createFileSystem2(new URI(this.dir), conf);
		absolutePath = new Path(this.dir, createFilename(prefixname));
		fs = absolutePath.getFileSystem(conf);
		if (fs.mkdirs(absolutePath)) {
			LogWriter.addLog("INFO","create data temp directory successfully "
					+ absolutePath.toString());
			
		} else {
			LogWriter.addLog("ERROR","Failed to mkdir " + absolutePath.toString());
			throw new WormholeException(JobStatus.READ_FAILED.getStatus());
		}
		return absolutePath.toString();
	}
	
	private void closeAll() {
		try {
			IOUtils.closeStream(fs);
		} catch (Exception e) {
			LogWriter.addLog("ERROR",String.format(
					"HdfsWriter closing filesystem failed: %s,%s",
					e.getMessage(), e.getCause()));
		}
	}

	private boolean existIncompleteFile(String dir) {
		try {
			fs = DFSUtils.createFileSystem(new URI(dir),
					DFSUtils.getConf(dir, null));
			for (FileStatus status : fs.listStatus(new Path(dir))) {
				if (status.getPath().getName().endsWith(".tmp")) {
					return true;
				}
			}
		} catch (Exception e) {
			LogWriter.addLog("ERROR",String.format(
					"HdfsWriter Init file system failed:%s,%s", e.getMessage(),
					e.getCause()));
		} finally {
			closeAll();
		}
		return false;
	}

	private void deleteFilesOnHdfs(String dir, String prefix, String suffix) {
		try {
			fs = DFSUtils.createFileSystem2(new URI(dir),
					DFSUtils.getConf2(dir, null));

			if (concurrency == 1) {
				String file = dir + "/" + prefix;

				if (!StringUtils.isEmpty(suffix))
					file = file + "." + suffix;

				DFSUtils.deleteFile(fs, new Path(file), true);

				if (lzoCompressed) {
					file = file + ".index";
					if (fs.exists(new Path(file))) {
						DFSUtils.deleteFile(fs, new Path(file), true);
					}
				}
			} else {
				DFSUtils.deleteFiles(fs, new Path(dir + "/" + prefix + "-*"),
						true, true);
			}
		} catch (Exception e) {
			LogWriter.addLog("ERROR",String.format(
					"HdfsWriter Init file system failed:%s,%s", e.getMessage(),
					e.getCause()));
		} finally {
			closeAll();
		}
	}
}
